

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.List;

import com.amazonaws.AmazonClientException;
import com.amazonaws.AmazonServiceException;
import com.amazonaws.auth.PropertiesCredentials;
import com.amazonaws.services.s3.AmazonS3;
import com.amazonaws.services.s3.AmazonS3Client;
import com.amazonaws.services.s3.model.ListObjectsRequest;
import com.amazonaws.services.s3.model.ObjectListing;
import com.amazonaws.services.s3.model.S3Object;
import com.amazonaws.services.s3.model.S3ObjectSummary;
import com.amazonaws.services.simpledb.AmazonSimpleDB;
import com.amazonaws.services.simpledb.AmazonSimpleDBClient;
import com.amazonaws.services.simpledb.model.CreateDomainRequest;
import com.amazonaws.services.simpledb.model.DeleteDomainRequest;
import com.amazonaws.services.simpledb.model.PutAttributesRequest;
import com.amazonaws.services.simpledb.model.ReplaceableAttribute;
import com.amazonaws.services.sqs.AmazonSQS;
import com.amazonaws.services.sqs.AmazonSQSClient;
import com.amazonaws.services.sqs.model.ChangeMessageVisibilityRequest;
import com.amazonaws.services.sqs.model.CreateQueueRequest;
import com.amazonaws.services.sqs.model.Message;
import com.amazonaws.services.sqs.model.ReceiveMessageRequest;
import com.amazonaws.services.sqs.model.SendMessageRequest;

public class ClientPdfLaucher {
	
	public static void main(String[] args) throws Exception {
        /*
         * Important: Be sure to fill in your AWS access credentials in the
         *            AwsCredentials.properties file before you try to run this
         *            sample.
         * http://aws.amazon.com/security-credentials
         */
        AmazonS3 s3 = new AmazonS3Client(new PropertiesCredentials(
        		ClientPdfLaucher.class.getResourceAsStream("AwsCredentials.properties")));
        
        AmazonSQS sqs = new AmazonSQSClient(new PropertiesCredentials(
        		ClientPdfLaucher.class.getResourceAsStream("AwsCredentials.properties")));
        
        AmazonSimpleDB sdb = new AmazonSimpleDBClient(new PropertiesCredentials(
        		ClientPdfLaucher.class.getResourceAsStream("AwsCredentials.properties")));

        System.out.println("===========================================");
        System.out.println("Client PDF Launcher");
        System.out.println("===========================================\n");

        try {
        	
        	/* delete pdfparagraphbucket */
        	ObjectListing objectListing = s3
				.listObjects(new ListObjectsRequest()
					.withBucketName("pdfparagraphbucket"));
			for (S3ObjectSummary objectSummary : objectListing.getObjectSummaries())
				s3.deleteObject(objectSummary.getBucketName(),objectSummary.getKey());
        	
        	/* delete plagiarismbucket */
			objectListing = s3
				.listObjects(new ListObjectsRequest()
					.withBucketName("plagiarismbucket"));
			for (S3ObjectSummary objectSummary : objectListing.getObjectSummaries())
				s3.deleteObject(objectSummary.getBucketName(),objectSummary.getKey());
        	
        	/* Create queue */
            System.out.println("Creating a new SQS queue called PdfQueue.\n");
            CreateQueueRequest createQueueRequest = new CreateQueueRequest("PdfQueue");
            String pdfQueueUrl = sqs.createQueue(createQueueRequest).getQueueUrl();
        	
            /*
             * Amazon S3 bucket names are globally unique,
             * so once a bucket name has been taken by any user, you can't create
             * another bucket with that same name.
             */            
            String pdfBucket = "exercise2.ws2011";
        	System.out.println("Accessing bucket exercise2.ws2011.\n");
			objectListing = s3
					.listObjects(new ListObjectsRequest()
							.withBucketName(pdfBucket));
			for (S3ObjectSummary objectSummary : objectListing.getObjectSummaries()) {
				// Send a message
				System.out.println("Sending message to PdfQueue: " + pdfBucket +";"+objectSummary.getKey()+"\n");
	            sqs.sendMessage(new SendMessageRequest(pdfQueueUrl, pdfBucket+";"+objectSummary.getKey()));
			}
			
			/* wait until queue is empty */
			System.out.println("Waiting until queue is empty.\n");
			
			List<Message> messages = sqs.receiveMessage(new ReceiveMessageRequest(
					pdfQueueUrl)).getMessages();
			while(messages.size()>0)
			{
				for(Message message : messages)
					sqs.changeMessageVisibility(new ChangeMessageVisibilityRequest(pdfQueueUrl, 
							message.getReceiptHandle(), 0) );
				
				Thread.sleep(2000);	
				messages = sqs.receiveMessage(new ReceiveMessageRequest(
						pdfQueueUrl)).getMessages();
			}
			
			/* delete plagiarismbucket */
			objectListing = s3
				.listObjects(new ListObjectsRequest()
					.withBucketName("plagiarismbucket"));
			for (S3ObjectSummary objectSummary : objectListing.getObjectSummaries())
				s3.deleteObject(objectSummary.getBucketName(),objectSummary.getKey());
			
			/* map reduce */
			System.out.println("Starting elastic map reduce.\n");
			new ElasticMapReduceProcess().start();
			System.out.println("Stopping elastic map reduce.\n");
			
			/* save top words to simpledb */
        	String domain = "ParagraphsDomain";
        	sdb.deleteDomain(new DeleteDomainRequest(domain));
        	
        	// Create domain
            System.out.println("Creating domain called " + domain + ".\n");
            sdb.createDomain(new CreateDomainRequest(domain));
        	
			objectListing = s3
				.listObjects(new ListObjectsRequest()
					.withBucketName("plagiarismbucket"));
			int itemCounter = 0;
			for (S3ObjectSummary objectSummary : objectListing.getObjectSummaries())
			{
				if(objectSummary.getKey().startsWith("run0/"))
				{
					S3Object object = s3.getObject(objectSummary.getBucketName(), objectSummary.getKey());
					
					BufferedReader reader = new BufferedReader(new InputStreamReader(object.getObjectContent()));
					String line;
					while((line = reader.readLine()) != null)
					{
						String[] split = line.split(";");
						if(split.length > 2)
						{
							List<ReplaceableAttribute> attributes = new ArrayList<ReplaceableAttribute>();
							
							attributes.add(new ReplaceableAttribute("Pdf",split[0],true));
							attributes.add(new ReplaceableAttribute("ParagraphNo",split[1],true));
							for(int i=2; i < split.length-1; i++)
							{
								attributes.add(new ReplaceableAttribute("Top10Words",
										split[i],true));
							}
							System.out.println("Putting data into " + domain + " domain.\n");
				            sdb.putAttributes(new PutAttributesRequest(domain, "item_" + itemCounter++, attributes));
						}
					}
				}
			}
			
			/* sampleDB queries */
			SimpleDBQueries.findPlagiats(domain, "users", "web", "pages");		
			
        } catch (AmazonServiceException ase) {
            System.out.println("Caught an AmazonServiceException, which means your request made it "
                    + "to Amazon, but was rejected with an error response for some reason.");
            System.out.println("Error Message:    " + ase.getMessage());
            System.out.println("HTTP Status Code: " + ase.getStatusCode());
            System.out.println("AWS Error Code:   " + ase.getErrorCode());
            System.out.println("Error Type:       " + ase.getErrorType());
            System.out.println("Request ID:       " + ase.getRequestId());
        } catch (AmazonClientException ace) {
            System.out.println("Caught an AmazonClientException, which means the client encountered "
                    + "a serious internal problem while trying to communicate, "
                    + "such as not being able to access the network.");
            System.out.println("Error Message: " + ace.getMessage());
        }
    }
}
